package com.bigdata.exam

import org.apache.spark.rdd.RDD
import org.apache.spark.sql.types.{DataType, DoubleType, StringType, StructField, StructType}
import org.apache.spark.sql.{DataFrame, Row, SparkSession}

object ExamBaseResult {

  def main(args: Array[String]): Unit = {
    val spark: SparkSession = SparkSession.builder()
      .master("local[1]")
      .appName("baseScore")
      .getOrCreate()

    spark.sparkContext.setLogLevel("error")

    val schema: StructType = new StructType()
      .add("name", StringType)
      .add("score", DoubleType)

    val baseScore: DataFrame = spark
      .read
      .schema(schema)
      .format("csv")
      .csv("data/exam/base_score.csv")

    //    baseScore.printSchema()
    //    baseScore.show(100, false)

    val fullScore: Double = 27
    val guakeName = Array("林国涛", "王汉阳", "张彪", "彭良宇", "沈红阳", "朱昊天", "张祺阳", "汪旭", "孙清林", "蔡杰", "许默")
    val rdd: RDD[Row] = baseScore.rdd
    rdd.foreach(row => {
      val name: String = row.getString(0).replace(" ", "")
      val score: Double = row.getDouble(1)

      // 计算出勤率，得到得比例
      val realScore = score / fullScore
      // 通过比例计算得到 10分制度
      val defen = (adjust(realScore, 2, 100) * 10).toInt


      // 四十分制，得到平时成绩
      val pingshi = (realScore * 100).round

      if (guakeName.contains(name)) {
        println("挂科---" + name + "---" + score + "---" + realScore + "----得分：" + 2 + "----平时成绩：" + 20)
        println("别挣扎了，已经挂了~！！！")
      } else {
        println(name + "---" + score + "---" + realScore + "----得分：" + defen + "----平时成绩：" + pingshi)
        // 计算还要多少分才能及格
        println("期末成绩：" + ((60 - (pingshi * 0.4)) / 0.6).ceil + " --- 分才能及格~！！！")
      }

      println("--------------------------")

    })
  }


  def adjust(value: Double, wei: Int, beishu: Int): Double = {
    // 保留两位小数，取小数点第二位
    val bd = BigDecimal(value).setScale(wei, BigDecimal.RoundingMode.DOWN)
    val secondDecimalPlace = (bd * beishu).toInt % 10

    // 如果小数点第二位小于5
    if (secondDecimalPlace > 5) {
      BigDecimal(value).setScale(1, BigDecimal.RoundingMode.UP).doubleValue()
    } else {
      BigDecimal(value).setScale(1, BigDecimal.RoundingMode.DOWN).doubleValue()
    }
  }
}
